/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package org.just.dusin.ooii.component.standard;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;
import java.util.Properties;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
import org.apache.log4j.Logger;
import org.just.dusin.ooii.component.Sink;
import org.just.dusin.ooii.component.Source;


/**
 *
 * @author Administrator
 */
public class AMQListenerSource extends Source {

    private String brokerURL = "tcp://localhost:61616";
    private ConnectionFactory factory;
    private Connection connection;
    private Session session;
    private String queue="Sync.sql";
    @Override
    public void initComponent(Properties props) {
        factory = new ActiveMQConnectionFactory(brokerURL);
    	try{
            connection = factory.createConnection("admin","admin");
            connection.start();
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        }catch(JMSException e){
            
        }
    }

    @Override
    public void destroyComponent() throws Exception {
         if (connection != null) {
            connection.close();
        }
    }

    @Override
    public void run() {
        try{
    		Destination destination = getSession().createQueue(queue);
    		MessageConsumer messageConsumer = getSession().createConsumer(destination);
    		messageConsumer.setMessageListener(new Listener(sink));    	
        }catch(JMSException e){
            
        }
    }
    
    public Session getSession() {
        return session;
    }   
    
    static class Listener implements MessageListener {

	private String job;
        private Sink sink;
	
	public Listener(Sink sink) {
            this.sink = sink;
	}

	public void onMessage(Message message) {
		try {
			//do something here
                        TextMessage msg = (TextMessage)message;
			sink.process(msg.getText());
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}
}

